package com.echo.im.engin.mq;


import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import com.alibaba.fastjson2.TypeReference;
import com.echo.im.common.model.RecvInfo;
import com.echo.im.engin.netty.processor.MessageOutProcessor;
import com.echo.im.engin.netty.processor.factory.ProcessorFactory;
import com.echo.im.infrastructure.mq.event.BaseEvent;
import com.rabbitmq.client.Channel;
import lombok.RequiredArgsConstructor;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.rabbit.listener.api.ChannelAwareBatchMessageListener;
import org.springframework.stereotype.Component;

import java.nio.charset.StandardCharsets;
import java.util.ArrayList;
import java.util.List;

/**
 * 统一监听并消费所有推送消息
 * @author: echo
 * @time: 2024-08-21
 */
@Component
@Slf4j
@RequiredArgsConstructor
public class PushMessageReceiver  implements ChannelAwareBatchMessageListener {

    // 批量处理的最大消息数
    private static final int BATCH_SIZE = 200;

    private final ProcessorFactory processorFactory;

    @SneakyThrows
    @Override
    public void onMessageBatch(List<Message> messages, Channel channel) {

        List<Long> deliveryTags = new ArrayList<>();

        try {
            int num=0;
            // 遍历处理从队列中拉取到的每条消息
            for (Message message : messages) {
                String msg = new String(message.getBody(), StandardCharsets.UTF_8);

                // 将消息体转换为自定义的业务对象
                BaseEvent.EventMessage<RecvInfo<?>> eventMessage
                        = JSON.parseObject(msg, new TypeReference<BaseEvent.EventMessage<RecvInfo<?>>>(){}.getType());

                RecvInfo recvInfo = eventMessage.getData();

                MessageOutProcessor<RecvInfo<?>> processor = processorFactory.createOutProcessor(recvInfo.getCommand());

                // 获取 processor 的泛型类型
                Class<?> genericType = GenericTypeResolver.getGenericTypeArgument(processor.getClass(), 0);

                recvInfo.setData(((JSONObject)recvInfo.getData()).toJavaObject(genericType));

                processor.outProcess(recvInfo);

                // 收集消息的deliveryTag以便批量确认
                deliveryTags.add(message.getMessageProperties().getDeliveryTag());

                num++;
                // 如果已达到批量大小，立即进行数据库批量插入和批量ACK
                if (num == BATCH_SIZE) {
                    // 批量确认消息，使用最后一个deliveryTag进行批量确认
                    long lastDeliveryTag = deliveryTags.get(deliveryTags.size() - 1);
                    // true表示确认所有小于等于此deliveryTag的消息
                    channel.basicAck(lastDeliveryTag, true);
                    deliveryTags.clear();
                    num=0;
                    log.info("批量ACK确认 {} 条消息", num);
                }
            }

            // 处理完所有消息后，如果还有未插入和未ACK的消息，进行最后的批量操作
            if (num!=0) {
                // 批量确认消息，使用最后一个deliveryTag进行批量确认
                long lastDeliveryTag = deliveryTags.get(deliveryTags.size() - 1);
                // true表示确认所有小于等于此deliveryTag的消息
                channel.basicAck(lastDeliveryTag, true);
                deliveryTags.clear();
                log.info("批量ACK确认 {} 条消息", num);
            }

        } catch (Exception e) {
            e.printStackTrace();
            log.error("处理消息出现异常：{}", e.getMessage());
            // 如果出现异常，拒绝处理并重新投递这些消息
            for (Long deliveryTag : deliveryTags) {
                // 重新投递
                channel.basicNack(deliveryTag, false, false);
            }
        }
    }

}

